// Project name:JavaRedis
// Creator:muhong
// Date time:2022/12/14,10:40 下午
// Name:SingleThreadEventExecutor

package org.gmh.util.concurrent;

import org.omg.CORBA.portable.UnknownException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedHashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

/**
 * 单执行器--单线程模型，也就是说一个执行器和线程是一一对应的关系
 */
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {

    private final static Logger logger = LoggerFactory.getLogger(SingleThreadEventExecutor.class);

    static final int DEFAULT_MAX_PENDING_EXECUTOR_TASKS = 16;

    private static final int ST_NOT_STARTED = 1;
    private static final int ST_STARTED = 2;
    private static final int ST_SHUTTING_DOWN = 3;
    private static final int ST_SHUTDOWN = 4;
    private static final int ST_TERMINATED = 5;

    private static final Runnable WAKEUP_TASK = new Runnable() {
        @Override
        public void run() {

        }
    };

    private static final Runnable NOOP_TASK = () -> {
    };

    private static final AtomicIntegerFieldUpdater<SingleThreadEventExecutor> STATE_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(SingleThreadEventExecutor.class, "state");

    private final Queue<Runnable> taskQueue;

    private volatile int state = ST_NOT_STARTED;

    // 该executor中只有这一个线程
    private volatile Thread thread;

    private final Executor executor;
    private volatile boolean interrupted;

    private final Semaphore threadLock = new Semaphore(0);
    private Set<Runnable> shutdownHooks = new LinkedHashSet<>();
    private final boolean addTaskWakesUp;
    private final int maxPendingTasks;
    private final RejectedExceptionHandler rejectedExecutionHandler;

    private long lastExecutionTime;

    private volatile long gracefulShutdownQuietPeriod;
    private volatile long gracefulShutdownTimeout;
    private long gracefulShutdownStartTime;

    private final Promise<?> terminationFuture = new DefaultPromise<Void>();

    protected SingleThreadEventExecutor(EventExecutorGroup parent, ThreadFactory threadFactory,
                                        boolean addTaskWakesUp, int maxPendingTasks, RejectedExceptionHandler rejectedExceptionHandler) {
        this(parent, new ThreadPerTaskExecutor(threadFactory), addTaskWakesUp, maxPendingTasks, rejectedExceptionHandler);
    }

    protected SingleThreadEventExecutor(EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {
        this(parent, new ThreadPerTaskExecutor(threadFactory), addTaskWakesUp);
    }

    protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor, boolean addTaskWakesUp) {
        this(parent, executor, addTaskWakesUp, DEFAULT_MAX_PENDING_EXECUTOR_TASKS, RejectedExceptionHandlers.reject());
    }

    protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                        boolean addTaskWakesUp, int maxPendingTasks,
                                        RejectedExceptionHandler rejectedExecutionHandler) {
        super(parent);
        this.executor = executor;
        this.addTaskWakesUp = addTaskWakesUp;
        this.maxPendingTasks = maxPendingTasks;
        this.rejectedExecutionHandler = rejectedExecutionHandler;
        taskQueue = newTaskQueue(maxPendingTasks);

    }

    protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
        return new LinkedBlockingQueue<>(maxPendingTasks);
    }

    protected void interruptThread() {
        Thread currentThread = thread;
        if (currentThread == null) {
            interrupted = true;
        } else {
            currentThread.interrupt();
        }
    }

    protected Runnable pollTask() {
        assert inEventLoop();
        return pollTaskFrom(taskQueue);
    }

    @Override
    public boolean inEventLoop(Thread thread) {
        return thread == this.thread;
    }

    protected static Runnable pollTaskFrom(Queue<Runnable> taskQueue) {
        for (; ; ) {
            Runnable task = taskQueue.poll();
            if (task == WAKEUP_TASK) {
                continue;
            }
            return task;
        }
    }

    protected final void reject(Runnable task) {
        rejectedExecutionHandler.rejected(task, this);
    }

    protected void addTask(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (!offerTask(task)) {
            reject(task);
        }
    }

    public int pendingTasks() {
        return taskQueue.size();
    }

    final boolean offerTask(Runnable task) {
        if (isShutdown()) {
            reject();
        }
        return taskQueue.offer(task);
    }

    protected Runnable taskTask() {
        assert inEventLoop();
        if (!(taskQueue instanceof LinkedBlockingQueue)) {
            throw new UnsupportedOperationException();
        }
        // TODO 定时任务的处理

        return pollTask();
    }

    protected static void reject() {
        throw new RejectedExecutionException("event executor terminated");
    }

    protected Runnable peekTask() {
        assert inEventLoop();
        return taskQueue.peek();
    }

    protected boolean hasTasks() {
        assert inEventLoop();
        return !taskQueue.isEmpty();
    }

    protected boolean removeTask(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        return taskQueue.remove(task);
    }

    protected boolean runAllTask() {
        logger.debug("start to run all task:" + taskQueue.size());
        assert inEventLoop();
        boolean fetchedAll;
        boolean atLeastOne = false;

        // TODO 暂时先不处理定时任务
        if (runAllTaskFrom(taskQueue)) {
            atLeastOne = true;
        }
        afterRunningAllTasks();
        return atLeastOne;
    }

    protected boolean runAllTask(long timeoutNanos) {
        // TODO 先从调度任务里面pool待执行的任务到任务队列中
        Runnable task = pollTask();
        if (task == null) {
            afterRunningAllTasks();
            return false;
        }
        // 这里的时钟netty有优化，使用了内部的时钟来计算
        final long deadline = timeoutNanos > 0 ? getCurrentTimeNanos() + timeoutNanos : 0;
        long runTask = 0;
        long lastExecutionTime;
        for (; ; ) {
            safeExecute(task);
            runTask++;
            if ((runTask & 0x3F) == 0) {
                lastExecutionTime = getCurrentTimeNanos();
                if (lastExecutionTime >= deadline) {
                    break;
                }
            }
            task = pollTask();
            if (task == null) {
                lastExecutionTime = getCurrentTimeNanos();
                break;
            }
        }
        afterRunningAllTasks();
        this.lastExecutionTime = lastExecutionTime;
        return true;
    }

    private long getCurrentTimeNanos() {
        return System.nanoTime();
    }

    protected void afterRunningAllTasks() {

    }

    protected final boolean runAllTaskFrom(Queue<Runnable> taskQueue) {
        Runnable task = pollTaskFrom(taskQueue);
        if (task == null) {
            return false;
        }
        for (; ; ) {
            safeExecute(task);
            task = pollTaskFrom(taskQueue);
            if (task == null) {
                return true;
            }
        }
    }

    /**
     * 具体执行器实现该方法
     */
    protected abstract void run();

    protected long delayNanos(long currentTimeNanos) {
        return SCHEDULE_PURGE_INTERVAL;
    }

    private static final long SCHEDULE_PURGE_INTERVAL = TimeUnit.SECONDS.toNanos(1);

    private void startThread() {
        logger.info("judge whether to start a new thread int executor:{},{}", this, state);
        if (state == ST_NOT_STARTED) {
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                try {
                    doStartThread();
                } catch (Throwable cause) {
                    STATE_UPDATER.set(this, ST_NOT_STARTED);
                    throw new UnknownException(cause);
                }
            }
        }
    }

    private void doStartThread() { //
        logger.info("create a new thread int executor");
        assert thread == null;
        executor.execute(new Runnable() { // executor当前是ThreadPerTaskExecutor
            @Override
            public void run() {
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }
                boolean success = false;
                try {
                    SingleThreadEventExecutor.this.run(); // 在线程中启动执行器的run方法
                    success = true;
                } catch (Throwable cause) {
                    logger.warn("unexpected exception from an event executor:", cause);
                } finally {
                    for (; ; ) {
                        int oldState = state;
                        if (oldState >= ST_SHUTTING_DOWN
                                || STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
                            break;
                        }
                    }
                    //
                    try {
                        for (; ; ) {
                            if (confirmShutdown()) {
                                break;
                            }
                        }

                    } finally {
                        try {
                            cleanup();
                        } finally {
                            STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
                            threadLock.release();
                            if (!taskQueue.isEmpty()) {
                                logger.warn("an event executor terminated with non-empty task queue");
                            }
                        }
                    }
                }
            }
        });
    }

    @Override
    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        boolean inEventLoop = inEventLoop();
        addTask(task);
        // 下面的逻辑保证当前eventloop中只有一个线程
        if (!inEventLoop) {
            startThread(); // 当前executor没有在线程中运行
            if (isShutdown() && removeTask(task)) {
                reject();
            }
        }

        if (!addTaskWakesUp && wakesUpForTask(task)) {
            logger.info("wakeup the selector from blocking the loop will run task from queue");
            wakeup(inEventLoop);
        }
    }

    protected boolean wakesUpForTask(Runnable task) {
        return true;
    }

    protected void wakeup(boolean inEventLoop) {
        if (!inEventLoop && state == ST_SHUTTING_DOWN) {
            taskQueue.offer(WAKEUP_TASK);
        }
    }

    protected boolean confirmShutdown() {
        return true;
    }

    protected void cleanup() {

    }

    @Override
    public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
        if (quietPeriod < 0) {
            throw new IllegalArgumentException("quietPeriod: " + quietPeriod + " (expected >= 0)");
        }
        if (timeout < quietPeriod) {
            throw new IllegalArgumentException("timeout: " + timeout + " (expected >= quietPeriod)");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }
        if (isShuttingDown()) {
            return terminationFuture();
        }
        boolean inEventLoop = inEventLoop();
        boolean wakeup;
        int oldState;
        for (; ; ) {
            if (isShuttingDown()) {
                return terminationFuture();
            }
            int newState;
            wakeup = true;
            oldState = state;
            if (inEventLoop) {
                newState = ST_SHUTTING_DOWN;
            } else {
                switch (oldState) {
                    case ST_NOT_STARTED:
                    case ST_STARTED:
                        newState = ST_SHUTTING_DOWN;
                        break;
                    default:
                        newState = oldState;
                        wakeup = false;
                }
            }
            if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {
                break;
            }
        }
        gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
        gracefulShutdownTimeout = unit.toNanos(timeout);

        if (oldState == ST_NOT_STARTED) {
            try {
                doStartThread();
            } catch (Throwable throwable) {
                STATE_UPDATER.set(this, ST_TERMINATED);
                terminationFuture.tryFailure(throwable);
                if (!(throwable instanceof Exception)) {
                    // throw exception
                    return terminationFuture;
                }
            }
        }

        if (wakeup) {
            wakeup(inEventLoop);
        }
        return terminationFuture();
    }

    @Override
    public Future<?> terminationFuture() {
        return terminationFuture;
    }

    @Override
    public boolean isShuttingDown() {
        return state >= ST_SHUTTING_DOWN;
    }
}
